package zone

import (
	c "context"
	"github.com/Unixeno/TheMoon/common/zone"
	"github.com/Unixeno/TheMoon/logging"
	"github.com/Unixeno/TheMoon/model"
	"github.com/Unixeno/TheMoon/moon/core/context"
	"github.com/Unixeno/TheMoon/moon/core/module"
	"github.com/Unixeno/TheMoon/moon/zone/network"
	"github.com/Unixeno/TheMoon/moon/zone/storage"
	"github.com/Unixeno/TheMoon/pool"
	message "github.com/Unixeno/TheMoon/radio/protos"
	zf "github.com/Unixeno/TheMoon/zonefile"
	"github.com/davecgh/go-spew/spew"
	"strings"
	"time"
)

const (
	defaultReportErrorInterval = 3
	ModuleName                 = "zone"
	RemoteModule               = "zoner"
)

var logger = logging.GetLogger(ModuleName)

type Module struct {
	ctx      c.Context
	cancel   c.CancelFunc
	reporter *postman
}

func GetModule() module.Module {
	zoneModule := &Module{}

	return zoneModule
}

func (m *Module) Init() {
	logger.Debug("init zone")
	m.ctx, m.cancel = context.NewContext()
	context.RegisterModule(ModuleName)
	storage.InitLocalStorage()
	network.InitNetwork()

	m.reporter = newPostman()

	initIslands(m.ctx)
	initLocalDeployers()
}

func (m *Module) Start() {
	m.restart()
	go m.handle()
	go m.autoSync()
}

func (m *Module) restart() {
	// 尝试重启所有的 zone
	zones := storage.ListZone()
	logger.Infof("found %d zones, recovering...", len(zones))

	for zoneID, _ := range zones {
		// 重新部署
		m.deploy(zoneID)
	}
}

func (m *Module) autoSync() {
	m.sendSync(nil) // 启动时强制拉取所有的zone列表
	for {
		select {
		case <-context.Done():
			return
		case <-time.After(time.Second * 15):
		}
		m.sendSync(storage.ListZone())
	}
}

func (m *Module) sendSync(source map[string]storage.LocalZoneInfo) {
	syncMessage := message.ZoneSyncMessage{}
	if source != nil {
		for zoneID, val := range source {
			syncMessage.Message = append(syncMessage.Message, &message.ZoneSyncSingle{
				ZoneID:         zoneID,
				CurrentVersion: val.Version,
				State:          int32(val.State),
			})
		}
	}
	buffer, _ := syncMessage.Marshal()

	msg := model.NewMessage(ModuleName, RemoteModule)
	msg.Content = buffer

	err := context.SendMessage(msg)

	if err != nil {
		logger.WithError(err).Error("failed to send sync at module start up")
	}
}

func (m *Module) handle() {
	logger.Debug("zone handle started")
	for {
		msg, err := context.RecvMessage(ModuleName)
		if err != nil {
			logger.WithError(err).Info("exit")
			return
		}
		data, ok := msg.Content.([]byte)
		if !ok {
			logger.Error("unknown message type")
			continue
		}

		switch msg.Operation {
		case zone.OpSync:
			syncMsg := &message.ZoneSyncMessage{}
			if err := syncMsg.Unmarshal(data); err != nil {
				logger.WithError(err).WithField("l", len(data)).Error("invalid message content at sync")
				continue
			}
			m.handleSync(syncMsg)
		case zone.OpLeave:
			logger.Debug("leave from zone")
			leaveMsg := &message.ZoneLeave{}
			if err := leaveMsg.Unmarshal(data); err != nil {
				logger.WithError(err).WithField("l", len(data)).Error("invalid message content at leave")
				continue
			}
			m.handleLeave(leaveMsg)
		case zone.OpFetch:
			fetchMsg := &message.ZoneFetch{}
			if err := fetchMsg.Unmarshal(data); err != nil {
				logger.WithError(err).Error("invalid message content at fetch")
				continue
			}
			d := loadOrCreateDeployer(fetchMsg.ZoneID)
			// 尝试上锁 tryLock return true 时 上锁成功
			// 否则当前 deployer 正在 fetch->handleFetch->deploy 流程之间
			if !d.tryLock() {
				continue
			}
			m.handleFetch(fetchMsg)
		case zone.OpGetContainers:
			m.handleGetContainers(&msg)
		default:
			logger.Error("unknown message operation")
		}
	}
}

func (m *Module) handleLeave(msg *message.ZoneLeave) {
	logger.Debug("handle leave")
	zoneID := msg.ZoneID
	d, ok := getDeployer(zoneID)
	if !ok {
		return
	}
	ctx, _ := context.NewContext()
	z, ok := storage.GetZone(zoneID)
	if !ok {
		return
	}
	// 执行 cleanTask
	zonefile, _ := zf.NewZoneFile(z.Rules)
	d.clean(ctx, zonefile)
	// 回收引用计数 <= 0 的卷和网络
	collector.collect()
	// 删除 deployer
	d.delete()
}

// plant 和 moon 之间的
func (m *Module) handleGetContainers(msg *model.Message) {
	logger.Debug("handle get moon containers...")
	ctx, _ := context.NewContext()
	list, err := containerList(ctx)
	moonContainerList := message.MoonContainerList{}
	if err != nil {
		moonContainerList.ErrorMsg = err.Error()
	} else {
		containers := make([]*message.MoonContainer, 0, len(list))
		for _, c := range list {
			temp := strings.Split(c.Name, "_")
			if len(temp) < 2 {
				continue
			}
			zoneID := temp[0]
			containers = append(containers, &message.MoonContainer{
				Name:   c.Name,
				ID:     c.Id,
				State:  c.State,
				Image:  c.Image,
				ZoneID: zoneID,
			})
		}
		moonContainerList.Containers = containers
	}
	// 发送消息
	reply := msg.CreateReplay()
	buf := pool.Fixed64KPool.Get().([]byte)
	defer pool.Fixed64KPool.Put(buf)
	n, err := moonContainerList.MarshalTo(buf)
	if err != nil {
		logger.WithError(err).Error()
		return
	}
	reply.Content = buf[:n]
	_ = context.SendMessage(reply)
}

func (m *Module) handleSync(msg *message.ZoneSyncMessage) {
	logger.Debug("handle sync")
	for _, val := range msg.Message {
		if z, ok := storage.GetZone(val.ZoneID); !ok {
			// 新的zone，必须要发起fetch请求来获取对应版本的zonefile
			// 创建一个默认的zone条目
			storage.PutZone(val.ZoneID, storage.LocalZoneInfo{
				Rules:   "",
				Version: 0,
				State:   zone.ZoneMoonCreated,
			})
			go m.fetch(val.ZoneID, val.CurrentVersion)
		} else if z.Version != val.CurrentVersion {
			// 本地落后于远程，需求发起sync
			logger.WithField("local", z.Version).WithField("remote", val.CurrentVersion).Debug("local version is outdated")
			go m.fetch(val.ZoneID, val.CurrentVersion)
		} else {
			// 到这里应该剩下的都是无需处理的zone
			logger.WithField("zoneID", val.ZoneID).Debug("synced")
		}
	}
}

func (m *Module) handleFetch(msg *message.ZoneFetch) {
	logger.Debug("handle fetch")
	if msg.Zonefile == "" && msg.Version != 0 {
		// zonefile内容不可为空，但版本号为0时除外
		storage.PutZone(msg.ZoneID, storage.LocalZoneInfo{Version: msg.Version, Rules: msg.Zonefile, State: zone.ZoneMoonCreated})
		logger.WithField("zoneID", msg.ZoneID).WithField("v", msg.Version).Warning("invalid zonefile, ignore...")
		return
	}
	storage.PutZone(msg.ZoneID, storage.LocalZoneInfo{
		Version: msg.Version,
		Rules:   msg.Zonefile,
		State:   zone.ZoneMoonSynced,
	})
	m.deploy(msg.ZoneID)
}

func (m *Module) fetch(zoneID string, version int32) {
	logger.WithField("zoneID", zoneID).WithField("v", version).Debug("fetch zonefile")
	d := loadOrCreateDeployer(zoneID)
	d.lock()

	fetchMessage := message.ZoneFetch{
		ZoneID:  zoneID,
		Version: version,
	}

	buf := pool.Fixed64KPool.Get().([]byte)
	defer pool.Fixed64KPool.Put(buf)
	n, err := fetchMessage.MarshalTo(buf)
	if err != nil {
		logger.WithError(err).Error()
		return
	}

	msg := model.NewMessage(ModuleName, RemoteModule)
	msg.Operation = zone.OpFetch
	msg.Content = buf[:n]
	// 使用同步发送来保证获得响应
	// todo: 错误时应该增加重试机制
	ret, err := context.SyncSendMessage(msg, time.Second*10)
	if err != nil {
		// 出错时先回滚状态
		logger.WithError(err).Error()
		storage.PutZone(zoneID, storage.LocalZoneInfo{Version: version, State: zone.ZoneMoonCreated})
		return
	}
	err = fetchMessage.Unmarshal(ret.Content.([]byte))
	if err != nil {
		logger.WithError(err).Error()
		storage.PutZone(zoneID, storage.LocalZoneInfo{Version: version, State: zone.ZoneMoonCreated})
		return
	}
	m.handleFetch(&fetchMessage)
}

func (m *Module) deploy(zoneID string) {
	logger.Debug(zoneID, "deploying...")

	z, _ := storage.GetZone(zoneID)
	// 检查状态
	if z.State == zone.ZoneMoonCreated || z.State == zone.ZoneMoonError {
		logger.WithField("zoneID", zoneID).WithField("state", z.State).Debug("ignore deploying")
		return
	}

	d := loadOrCreateDeployer(zoneID)
	ctx, _ := context.NewContext()

	// 验证 zonefile 的合法性
	logger.WithField("zoneID", zoneID).Debug("verifying zonefile")
	zonefile, err := zf.NewZoneFile(z.Rules)
	if err != nil {
		logger.WithError(err).Error("invalid zonefile format, ignore...")
		m.triggerError(zoneID, err, d.version)
		return
	}
	spew.Dump(zonefile)

	if z.State == zone.ZoneMoonSynced {
		// 刚得到新的 zonefile 的时候需要清理旧的部署
		if err := d.clean(ctx, zonefile); err != nil {
			m.triggerError(zoneID, err, d.version)
			return
		}
		logger.WithField("zoneID", zoneID).Debug("first deploy, execute init task")
		err = zonefile.ExecuteInitTask(ctx, d)
		if err != nil {
			logger.WithError(err).WithField("zoneID", zoneID).Error("failed to execute init task")
			m.triggerError(zoneID, err, z.Version)
			d.clean(ctx, zonefile)
			return
		}
	}

	// 部署 deploy task
	err = zonefile.ExecuteDeployTask(ctx, d)
	if err != nil {
		logger.WithError(err).WithField("zoneID", zoneID).Error("failed to deploy zone")
		m.triggerError(zoneID, err, z.Version)
		d.clean(ctx, zonefile)
		return
	}

	// 部署完毕后 回收卷
	logger.Debug("collect 0 referring count volumes and networks")
	collector.collect()
	logger.WithField("zoneID", zoneID).Info("deploy success")
	storage.UpdateZoneState(zoneID, zone.ZoneMoonDeployed)

	// 整个流程解锁
	d.unlock()
}

func (m *Module) triggerError(zoneID string, err error, version int32) {
	storage.UpdateZoneState(zoneID, zone.ZoneMoonError)
	m.reporter.TriggerEvent(message.Event_Error, "zone_"+zoneID, err.Error())
}
